package mq_service

import (
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
	"time"
)

type MqServer struct {
	conn *amqp.Connection
	ch *amqp.Channel
}

func NewMqServer() (*MqServer, error) {
	uri := "amqp://admin:admin001@localhost:5672/report"
	conn, err := amqp.Dial(uri)
	if err != nil {
		return nil, err
	}

	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	//ch.Qos(20, 0, false)

	mq := MqServer{
		conn: conn,
		ch: ch,
	}
	return &mq, nil
}

// SendMsg 发送消息
func (mq *MqServer) SendMsg(exchange, route string, msg []byte) error {
	return mq.ch.Publish(exchange, route, false, false, amqp.Publishing{
		ContentType: "application/json",
		Body: msg,
	})
}

func (mq *MqServer) SendMsgExp(exchange, route string, msg []byte, expire int) error {
	return mq.ch.Publish(exchange, route, false, false, amqp.Publishing{
		ContentType: "application/json",
		Body: msg,
		Expiration: fmt.Sprintf("%d", expire * 1000),
	})
}

// ReceiveMsg 接收消息
func (mq *MqServer) ReceiveMsg(queue string, handler func([]byte) error) {

	msg, err := mq.ch.Consume(queue, "", false, false, false, false, nil)
	if err != nil {
		log.Println("Failed to receive message err:", err.Error())
		return
	}
	quit := make(chan bool)
	doing := make(chan bool, 1)
	go func(doing chan bool) {
		for d := range msg {
			doing <- true
			log.Printf("data:%+v", string(d.Body))
			err := handler(d.Body)
			if err != nil {
				log.Println("消息处理失败:", err.Error(), string(d.Body))
			}
			d.Ack(false)
			<- doing
		}
	}(doing)
	log.Println(" Waiting for messages. queue:", queue)
	go func(quit chan bool) {
		for {
			if mq.GetQueueLen(queue) == 0 && len(doing) == 0 {
				time.Sleep(time.Second * 30)
				if mq.GetQueueLen(queue) == 0 && len(doing) == 0 {
					quit <- true
				}
			}
			fmt.Println(queue, "队列检查")
			time.Sleep(time.Second * 3)
		}

	}(quit)

	<-quit
}

func (mq *MqServer) MulReceiveMsg(queue string, handler func([]byte) error, quantity int) {
	for i := 0; i < quantity; i++ {
		go func(i int) {
			consumer := fmt.Sprintf("%s_consumer_%d", queue, i)
			//mq.ch.Qos(20, 0, false)
			msg, err := mq.ch.Consume(queue, consumer, false, false, false, false, nil)
			if err != nil {
				fmt.Println(err.Error())
				//logger.Errorf("Failed to receive message err:%s", err.Error())
				return
			}
			fmt.Println("consumer-", i)
			for d := range msg {
				//logger.Printf("data:%v", string(d.Body))
				log.Println(consumer)
				err := handler(d.Body)
				if err != nil {
					log.Println("消息处理失败:", err.Error(), string(d.Body))
				}
				d.Ack(false)
			}
		}(i)
	}

	//挂起
	forever := make(chan bool)
	<-forever
}

func (mq *MqServer) GetQueueLen(queueName string) int {
	queue, _ := mq.ch.QueueDeclarePassive(queueName, true, false, false, false, nil)
	return queue.Messages
}

func (mq *MqServer) Shutdown() {

	if !mq.ch.IsClosed() {
		mq.ch.Close()
	}

	if !mq.conn.IsClosed() {
		mq.conn.Close()
	}
}

